package com.yoyo.schedule.task_manager.processor;

import com.yoyo.schedule.TaskItemAndParameter;
import com.yoyo.schedule.task_get.MultipleTask;
import com.yoyo.schedule.task_get.SingleTask;
import com.yoyo.schedule.task_get.TaskListGetter;
import com.yoyo.schedule.task_manager.AbstractScheduleManager;
import com.yoyo.schedule.task_manager.Statistics;
import com.yoyo.schedule.task_manager.TaskType;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
public class SleepProcessor<Task> implements IProcessor, Runnable {
    final ObjectLock objectLock = new ObjectLock();
    @Setter
    @Getter
    protected AbstractScheduleManager abstractScheduleManager;
    @Setter
    @Getter
    protected TaskListGetter<Task> taskListGetter;
    @Setter
    @Getter
    protected List<Task> taskList = new CopyOnWriteArrayList<Task>();
    @Setter
    @Getter
    List<Thread> threadList = new CopyOnWriteArrayList<Thread>();
    @Setter
    @Getter
    TaskType taskType;
    @Setter
    @Getter
    boolean multipleTask = false;

    @Setter
    @Getter
    boolean stopSchedule = false;
    @Setter
    @Getter
    boolean isSleeping = false;
    @Setter
    @Getter
    Statistics statistics;


    public SleepProcessor(//
                          AbstractScheduleManager abstractScheduleManager,//
                          TaskListGetter<Task> taskListGetter,//
                          Statistics statistics) throws Exception//
    {
        this.abstractScheduleManager = abstractScheduleManager;
        this.taskType = this.abstractScheduleManager.getTaskType();
        //
        this.taskListGetter = taskListGetter;
        if (this.taskListGetter instanceof SingleTask<?>) {
            if (taskType.getMaxExecuteNumber() > 1) {
                taskType.setMaxExecuteNumber(1);//自动修正
            }
            multipleTask = false;
        } else {
            multipleTask = true;
        }
        if (taskType.getFetchNumber() >= taskType.getThreadNumber() * 10) {
            log.warn("fetchnum >= threadnum*10 ");
        }
        //
        this.statistics = statistics;
        //
        for (int i = 0; i < taskType.getThreadNumber(); i++) {
            this.startThread(i);
        }
    }


    private void startThread(int index) {
        Thread thread = new Thread(this);//共享信息
        threadList.add(thread);
        String threadName = //
                this.abstractScheduleManager.getTaskHandler().getTaskType() + "-"//
                        + this.abstractScheduleManager.getCurrentSerialNumber() + "-exe"//
                        + index;//
        thread.setName(threadName);
        thread.start();
    }

    //
    public boolean isSleeping() {
        return this.isSleeping;
    }

    public boolean taskListIsAllFinshed() {
        return this.taskList.size() == 0;
    }

    public void stopSchedule() throws Exception {

        this.stopSchedule = true;

        this.taskList.clear();
    }

    public void clearTaskList() {
        this.taskList.clear();
    }

    //
    public void run() {
        try {
            long startTime = 0;
            while (true) {
                this.objectLock.increaseThreadCount1();
                Object task;
                while (true) {
                    //while (true) do{this}
                    if (this.stopSchedule) {
                        //
                        this.objectLock.reduceThreadCount1();
                        this.objectLock.objectNotifyAll();
                        //
                        synchronized (this.threadList) {
                            this.threadList.remove(Thread.currentThread());
                            if (this.threadList.size() == 0) {
                                this.abstractScheduleManager.unRegisterTaskHandler();
                            }//all over
                        }
                        return;
                    }
                    //!stopSchedule

                    //load task from here
                    if (this.multipleTask == false) {
                        task = this.getTask();
                    } else {
                        task = this.getTasks();
                    }

                    if (task == null) {
                        break;
                    }

                    try {
                        startTime = abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis();
                        if (!this.multipleTask)//SingleTask
                        {
                            if (((SingleTask) this.taskListGetter).execute(task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                                addTaskSuccessNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            } else {
                                addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            }
                        } else {//MultipleTask
                            if (((MultipleTask) this.taskListGetter)
                                    .execute((Object[]) task, abstractScheduleManager.getTaskHandler().getOwnSign()) == true) {
                                addTaskSuccessNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            } else {
                                addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                        - startTime);
                            }
                        }
                    } catch (Throwable ex) {
                        if (!this.multipleTask) {
                            addFailNum(1, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis() - startTime);
                        } else {
                            addFailNum(((Object[]) task).length, abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()
                                    - startTime);
                        }
                        log.warn("Task :" + task + " ", ex);
                    }
                }//end while

                //has set(this.objectLock.increaseThreadCount1();)
                if (this.objectLock.reduceThreadCount1IfThreadCountNotEqual1AndReturnTheCurrentThreadIsNotLastThread()) {
                    this.objectLock.objectWait();
                } else {//current thread is the last thread

                    Thread.sleep(100);
                    startTime = this.abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis();

                    int taskSize = this.fetchTasks();
                    if (taskSize > 0) {
                        this.objectLock.objectNotifyAll();
                    } else {
                        if (this.stopSchedule == false && this.abstractScheduleManager.isContinueWhenHasData()) {
                            this.isSleeping = true;
                            Thread.sleep(this.abstractScheduleManager.getTaskType().getSleepTimeNoData());
                            this.isSleeping = false;

                        } else {
                            this.objectLock.objectNotifyAll();
                        }
                    }
                    this.objectLock.reduceThreadCount1();
                }
            }
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }
    }

    public synchronized Object getTask() {
        if (this.taskList.size() == 0) {
            return null;
        } else {
            return this.taskList.remove(0);//第一个
        }
    }

    public synchronized Object[] getTasks() {
        if (this.taskList.size() == 0) {
            return null;
        }
        int minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskList.size() > taskType.getMaxExecuteNumber() ? taskType.getMaxExecuteNumber() : taskList.size();
        Object[] taskList = null;
        if (!(minSizeBetweenTaskListSizeAndMaxExecuteNumber > 0)) {
            return taskList;
        } else {
            taskList = (Object[]) Array.newInstance(this.taskList.get(0).getClass(), minSizeBetweenTaskListSizeAndMaxExecuteNumber);
            for (int i = 0; i < minSizeBetweenTaskListSizeAndMaxExecuteNumber; i++) {
                taskList[i] = this.taskList.remove(0);//第一个
            }
            return taskList;
        }

    }

    /**
     * 获取任务
     *
     * @return 任务数量
     */
    protected int fetchTasks() {
        try {
            if (this.taskType.getSleepTimeInterval() > 0) {

                this.isSleeping = true;
                Thread.sleep(taskType.getSleepTimeInterval());
                this.isSleeping = false;

            }

            List<TaskItemAndParameter> taskItemAndParameterList = this.abstractScheduleManager.getTaskItemAndParameterList();
            if (taskItemAndParameterList.size() > 0) {
                List<TaskItemAndParameter> newTaskItemAndParameterList = new ArrayList<TaskItemAndParameter>();
                synchronized (taskItemAndParameterList) {
                    for (TaskItemAndParameter taskItemAndParameter : taskItemAndParameterList) {
                        newTaskItemAndParameterList.add(taskItemAndParameter);
                    }
                }
                List<Task> taskList = this.taskListGetter.getTaskList(//
                        taskType.getTaskParameter(),//
                        abstractScheduleManager.getTaskHandler().getOwnSign(),//
                        this.abstractScheduleManager.getTaskItemCount(), //
                        taskType.getFetchNumber(),//
                        newTaskItemAndParameterList//
                );//
                abstractScheduleManager.getTaskHandler().setLastFetchDataTime(new Timestamp(abstractScheduleManager.dataManager.zooKeeperCurrentTimeMillis()));
                if (taskList != null) {
                    this.taskList.addAll(taskList);
                }
            } else {
            }
            addTaskFetchNum(taskList.size());
            return this.taskList.size();
        } catch (Throwable ex) {
            log.error("Get tasks error.", ex);
        }
        return 0;
    }

    //
    public void addTaskFetchNum(long num) {

        this.statistics.addFetchDataCount(1);
        this.statistics.addTaskFetchNum(num);
    }

    public void addTaskSuccessNum(long num, long spendTime) {
        this.statistics.addDealDataSucess(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime) {
        this.statistics.addDealDataFail(num);
        this.statistics.addDealSpendTime(spendTime);
    }
}
